package com.atguigu.flink.chapter06.timer;

import com.atguigu.flink.bean.WaterSensor;
import com.atguigu.flink.function.WaterSensorMapFunction;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class TimerTest {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        //如果水位连续5s 超过 20 ，则发出红色预警
        //当第一个超过20 来的时候，注册定时器  5s后触发，如果后面来的都是>20的，则无需操作，一旦出现小于等于20的，则取消定时器
        env.socketTextStream("hadoop102",8888)
                .map(new WaterSensorMapFunction())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<WaterSensor>forMonotonousTimestamps()
                                .withTimestampAssigner((ws,ts) -> ws.getTs())
                )
                .keyBy(WaterSensor::getId)
                .process(new KeyedProcessFunction<String, WaterSensor, String>() {

                    @Override
                    public void onTimer(long timestamp,
                                        OnTimerContext ctx,
                                        Collector<String> out) throws Exception {
                        out.collect("水位连续5s超过20，触发红色预警...."  +timestamp);
                        isFirst20 = true;
                    }

                    boolean isFirst20 = true;

                    long timerTs;

                    @Override
                    public void processElement(WaterSensor value,
                                               Context ctx,
                                               Collector<String> out) throws Exception {
                        if (value.getVc() > 20 && isFirst20){
                            isFirst20 = false;
                            //注册一个5s后触发的定时器
                            timerTs = value.getTs() + 5000;
                            System.out.println("第一个大于20 的来了，注册定时器：" + timerTs);
                            ctx.timerService().registerProcessingTimeTimer(timerTs);
                        } else if (value.getVc() < 20) {
                            System.out.println("取消定时器：" + timerTs);
                            ctx.timerService().deleteProcessingTimeTimer(timerTs);
                            //把isFirst20 设置为true
                            isFirst20 = true;
                        }
                    }
                })
                .print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}



//public class TimerTest {
//    public static void main(String[] args) {
//        Configuration conf = new Configuration();
//        conf.setInteger("rest.port", 2000);
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
//        env.setParallelism(1);
//        //如果水位连续5s 都超过 20 ，则触发红色预警
//        //如果第一个20来的时候，注册定时器，5s后触发定时器，如果后面来的都超过20，则无需此操作，一旦水位大于等于20，则取消定时器
//        env.socketTextStream("hadoop102",8888)
//                .map(new WaterSensorMapFunction())
//                .assignTimestampsAndWatermarks(
//                        WatermarkStrategy
//                                .<WaterSensor>forMonotonousTimestamps()
//                                .withTimestampAssigner((ws,ts) -> ws.getTs())
//                )
//                .keyBy(WaterSensor::getId)
//                .process(new KeyedProcessFunction<String, WaterSensor, String>() {
//
//                    @Override
//                    public void onTimer(long timestamp,
//                                        OnTimerContext ctx,
//                                        Collector<String> out) throws Exception {
//                        out.collect("水位连续5s超过20，发出红色预警..." + timestamp);
//                        isFirst20 = true;
//                    }
//
//                    boolean isFirst20 = true;
//                    long timeTs;
//                    @Override
//                    public void processElement(WaterSensor value,
//                                               Context ctx,
//                                               Collector<String> out) throws Exception {
//                        if (value.getVc() > 20){
//                            isFirst20 = false;
//                            //注册一个定时器
//                            timeTs = value.getTs() + 5000;
//                            System.out.println("第一个20来的时候，注册定时器;"  +timeTs);
//                            ctx.timerService().registerProcessingTimeTimer(timeTs);
//                        } else if (value.getVc() < 20) {
//                            System.out.println("取消定时器" + timeTs);
//                            ctx.timerService().deleteProcessingTimeTimer(timeTs);
//                            //把isFist20 设置为true
//                            isFirst20 = true;
//                        }
//                    }
//                })
//                .print();
//
//
//        try {
//            env.execute();
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//    }
//}



//public class TimerTest {
//    public static void main(String[] args) {
//        Configuration conf = new Configuration();
//        conf.setInteger("rest.port", 2000);
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
//        env.setParallelism(1);
//        //如果水位连续5s都超过20，则发出红色预警
//        //当第一个20来的时候，注册定时器，5s后触发定时器，如果后面来的都是大于20的，则无需操作，一旦水位小于等于20，则取消定时器
//        env.socketTextStream("hadoop102",8888)
//                .map(new WaterSensorMapFunction())
//                .assignTimestampsAndWatermarks(
//                        WatermarkStrategy
//                                .<WaterSensor>forMonotonousTimestamps()
//                                .withTimestampAssigner((ws,ts) -> ws.getTs())
//                )
//                .keyBy(WaterSensor::getId)
//                .process(new KeyedProcessFunction<String, WaterSensor, String>() {
//
//                    @Override
//                    public void onTimer(long timestamp,
//                                        OnTimerContext ctx,
//                                        Collector<String> out) throws Exception {
//                        out.collect("水位连续5s超过20,发出红色预警...." + timestamp);
//                        isFirst20 = true;
//                    }
//
//                    boolean isFirst20 = true;
//                    long timerTs;
//                    @Override
//                    public void processElement(WaterSensor value,
//                                               Context ctx,
//                                               Collector<String> out) throws Exception {
//                        if (value.getVc() > 20){
//                            isFirst20 = false;
//                            //注册一个定时器
//                            timerTs = value.getTs() + 5000;
//                            System.out.println("第一个20来的时候，注册一个定时器："  + timerTs);
//                            ctx.timerService().registerProcessingTimeTimer(timerTs);
//                        }else if (value.getVc() < 20){
//                            System.out.println("取消定时器：" + timerTs);
//                            ctx.timerService().deleteProcessingTimeTimer(timerTs);
//                            //把isFirst20 设置为true
//                            isFirst20 = true;
//                        }
//                    }
//                })
//                .print();
//
//
//        try {
//            env.execute();
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//    }
//}